package openjdk.concurrent;

import java.util.Random;
import java.util.concurrent.*;

/**
 * 字面意思回环栅栏，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后，
 * CyclicBarrier可以被重用。我们暂且把这个状态就叫做barrier，当调用await()方法之后，线程就处于barrier了。
 * <p>
 * CyclicBarrier类位于java.util.concurrent包下，CyclicBarrier提供2个构造器：
 * public CyclicBarrier(int parties) { }
 * public CyclicBarrier(int parties, Runnable barrierAction) { }
 * <p>
 * 参数parties指让多少个线程或者任务等待至barrier状态；参数barrierAction为当这些线程都达到barrier状态时会执行的内容。
 * <p>
 * <p>
 * CyclicBarrier中最重要的方法就是await方法，它有2个重载版本：
 * <p>
 * public int await() throws InterruptedException, BrokenBarrierException { };
 * public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException { };
 * <p>
 * 第一个版本比较常用，用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；
 * 第二个版本是让这些线程等待至一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。
 * <p>
 * 在所有参与者都已经在此 barrier 上调用 await方法之前，将一直等待。如果当前线程不是将到达的最后一个线程，出于调度目的，将禁用它，
 * 且在发生以下情况之一前，该线程将一直处于休眠状态：
 * 最后一个线程到达；或者
 * 其他某个线程中断当前线程；或者
 * 其他某个线程中断另一个等待线程；或者
 * 其他某个线程在等待 barrier 时超时；或者
 * 其他某个线程在此 barrier 上调用 reset()。
 * 如果当前线程：
 * 在进入此方法时已经设置了该线程的中断状态；或者
 * 在等待时被中断
 * <p>
 * 则抛出 InterruptedException，并且清除当前线程的已中断状态。如果在线程处于等待状态时 barrier 被 reset()，
 * 或者在调用 await 时 barrier 被损坏，抑或任意一个线程正处于等待状态，则抛出 BrokenBarrierException 异常。
 * 如果任何线程在等待时被 中断，则其他所有等待线程都将抛出 BrokenBarrierException 异常，并将 barrier 置于损坏状态。
 * 如果当前线程是最后一个将要到达的线程，并且构造方法中提供了一个非空的屏障操作，则在允许其他线程继续运行之前，当前线程将运行该操作。
 * 如果在执行屏障操作过程中发生异常，则该异常将传播到当前线程中，并将 barrier 置于损坏状态。
 * <p>
 * <p>
 * 返回：
 * 到达的当前线程的索引，其中，索引 getParties() - 1 指示将到达的第一个线程，零指示最后一个到达的线程
 * 抛出：
 * InterruptedException - 如果当前线程在等待时被中断
 * BrokenBarrierException - 如果另一个 线程在当前线程等待时被中断或超时，或者重置了 barrier，或者在调用 await 时
 * barrier 被损坏，抑或由于异常而导致屏障操作（如果存在）失败。
 * <p>
 * <p>
 * <p>
 * <p>
 * <p>
 * 应用场景：
 * 一个大型的任务，常常需要分配好多子任务去执行，只有当所有子任务都执行完成时候，才能执行主任务，这时候可以选择CyclicBarrier
 * <p>
 * 示例：
 * CyclicBarrier 表示大家彼此等待，大家集合好后才开始出发，分散活动后又在i指定地点集合碰面，这就好比整个公司的人员利用周末时间集体郊游一样，
 * 先各自从家出发到公司集合后，再同时出发到公园游玩，在指定地点集合后再同时开始就餐……
 */
public class CyclicBarrierDemo1 {
    public static void main(String[] args) {
        //testWriteXXX();
        //parties();
        //barrierAction();
        //isBroken();
        runner();
    }


    // *********************************************
    //      屏障的基本用法
    // *********************************************


    /**
     * 此例输出 2 1 或者  1 2
     * 如果把new CyclicBarrier(2)修改成new CyclicBarrier(3)则主线程和子线程会永远等待，因为没有第三个线程执行await方法，
     * 即没有第三个线程到达屏障，所以之前到达屏障的两个线程都不会继续执行。
     */
    private static void parties() {
        CyclicBarrier c = new CyclicBarrier(2);

        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                }
                System.out.println(1);
            }
        }).start();

        try {
            c.await();
        } catch (Exception e) {
        }
        System.out.println(2);
    }


    /**
     * 优先执行barrierAction
     * <p>
     * 结果输出为 barrierAction arrived. 1 2
     */

    private static void barrierAction() {

        CyclicBarrier c = new CyclicBarrier(2, new A());

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {

                }
                System.out.println(1);
            }
        }).start();
        try {
            c.await();
        } catch (Exception e) {

        }
        System.out.println(2);
    }

    static class A implements Runnable {

        @Override
        public void run() {
            System.out.println("barrierAction arrived.");
        }
    }


    /**
     * CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。
     * isBroken方法用来知道阻塞的线程是否被中断。比如以下代码执行完之后会返回true。
     */
    private static void isBroken() {
        CyclicBarrier c = new CyclicBarrier(2);

        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {

                }
            }
        });

        thread.start();
        thread.interrupt();

        try {
            c.await();
        } catch (Exception e) {
            System.out.println(c.isBroken());
        }
    }


    /**
     * 赛跑时，等待所有人都准备好时，才起跑：
     */

    private static void runner() {
        //如果将参数改为4，但是下面只加入了3个选手，这永远等待下去
        //Waits until all parties have invoked await on this barrier.
        CyclicBarrier barrier = new CyclicBarrier(3);

        ExecutorService executor = Executors.newFixedThreadPool(3);
        executor.submit(new Thread(new Runner(barrier, "1号选手")));
        executor.submit(new Thread(new Runner(barrier, "2号选手")));
        executor.submit(new Thread(new Runner(barrier, "3号选手")));

        executor.shutdown();
    }

    static class Runner implements Runnable {
        // 一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)
        private CyclicBarrier barrier;

        private String name;

        public Runner(CyclicBarrier barrier, String name) {
            super();
            this.barrier = barrier;
            this.name = name;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(1000 * (new Random()).nextInt(8));
                System.out.println(name + " 准备好了...");
                // barrier的await方法，在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println(name + " 起跑！");
        }
    }


    /**
     * 假若有若干个线程都要进行写数据操作，并且只有所有线程都完成写数据操作之后，这些线程才能继续做后面的事情.
     */
    private static void testWrite() {
        int N = 4;
        CyclicBarrier barrier = new CyclicBarrier(N);
        for (int i = 0; i < N; i++)
            new WriterThread(barrier, 0).start();
    }


    /**
     * 从testWrite()输出结果可以看出，每个写入线程执行完写数据操作之后，就在等待其他线程写入操作完毕。
     * 当所有线程线程写入操作完毕之后，所有线程就继续进行后续的操作了。
     * 如果说想在所有线程写入操作完之后，进行额外的其他操作可以为CyclicBarrier提供Runnable参数：
     */
    private static void testWriteX() {

        int N = 4;
        CyclicBarrier barrier = new CyclicBarrier(N, new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程" + Thread.currentThread().getName());
            }
        });

        for (int i = 0; i < N; i++)
            new WriterThread(barrier, 0).start();
    }

    /**
     * 从testWriteX()结果可以看出，当四个线程都到达barrier状态后，会从四个线程中选择一个线程去执行Runnable。
     * 下面看一下为await指定时间的效果：
     */
    private static void testWriteXX() {

        int N = 4;
        CyclicBarrier barrier = new CyclicBarrier(N);

        for (int i = 0; i < N; i++) {
            if (i < N - 1)
                new WriterThread(barrier, 2000).start();
            else {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                new WriterThread(barrier, 2000).start();
            }
        }
    }


    /**
     * 上面的代码在main方法的for循环中，故意让最后一个线程启动延迟，因为在前面三个线程都达到barrier之后，
     * 等待了指定的时间发现第四个线程还没有达到barrier，就抛出异常并继续执行后面的任务。
     * <p>
     * CyclicBarrier是可以重用的，下面这个例子可以很好地说明此特性
     * <p>
     * <p>
     * 从执行结果可以看出，在初次的4个线程越过barrier状态后，又可以用来进行新一轮的使用。而CountDownLatch无法进行重复使用。
     */
    private static void testWriteXXX() {
        int N = 4;
        CyclicBarrier barrier = new CyclicBarrier(N);

        for (int i = 0; i < N; i++) {
            new WriterThread(barrier, 0).start();
        }

        try {
            Thread.sleep(25000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("CyclicBarrier重用");

        for (int i = 0; i < N; i++) {
            new WriterThread(barrier, 0).start();
        }
    }


    static class WriterThread extends Thread {
        private long timeout;
        private CyclicBarrier cyclicBarrier;

        public WriterThread(CyclicBarrier cyclicBarrier, long timeout) {
            this.cyclicBarrier = cyclicBarrier;
            this.timeout = timeout;
        }

        @Override
        public void run() {
            System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
            try {
                Thread.sleep(5000);      //以睡眠来模拟写入数据操作
                System.out.println("线程" + Thread.currentThread().getName() + "写入数据完毕，等待其他线程写入完毕");
                if (timeout > 1000) {
                    cyclicBarrier.await();
                } else {
                    try {
                        cyclicBarrier.await(timeout, TimeUnit.MILLISECONDS);
                    } catch (TimeoutException e) {
                        e.printStackTrace();
                    }
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("所有线程写入完毕，继续处理其他任务...");
        }

    }

}
